/*-------------------------------------------------------------------------
 * launcher.c
 *	   PostgreSQL 逻辑复制工作进程启动器
 *
 * Copyright (c) 2016-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/replication/logical/launcher.c
 *
 * NOTES
 *	  此模块包含逻辑复制工作进程启动器，
 *	  使用后台工作基础设施为每个启用的订阅启动
 *	  逻辑复制工作进程。
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/heapam.h"
#include "access/htup.h"
#include "access/htup_details.h"
#include "access/tableam.h"
#include "access/xact.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_subscription_rel.h"
#include "funcapi.h"
#include "libpq/pqsignal.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/bgworker.h"
#include "postmaster/fork_process.h"
#include "postmaster/interrupt.h"
#include "postmaster/postmaster.h"
#include "replication/logicallauncher.h"
#include "replication/logicalworker.h"
#include "replication/slot.h"
#include "replication/walreceiver.h"
#include "replication/worker_internal.h"
#include "storage/ipc.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/procsignal.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/pg_lsn.h"
#include "utils/ps_status.h"
#include "utils/snapmgr.h"
#include "utils/timeout.h"

/* 最大循环间的睡眠时间（3分钟） */
#define DEFAULT_NAPTIME_PER_CYCLE 180000L

int			max_logical_replication_workers = 4;
int			max_sync_workers_per_subscription = 2;

LogicalRepWorker *MyLogicalRepWorker = NULL;

typedef struct LogicalRepCtxStruct
{
	/* 监控进程。 */
	pid_t		launcher_pid;

	/* 后台工作者。 */
	LogicalRepWorker workers[FLEXIBLE_ARRAY_MEMBER];
} LogicalRepCtxStruct;

static LogicalRepCtxStruct *LogicalRepCtx;

static void fc_ApplyLauncherWakeup(void);
static void fc_logicalrep_launcher_onexit(int fc_code, Datum fc_arg);
static void fc_logicalrep_worker_onexit(int fc_code, Datum fc_arg);
static void fc_logicalrep_worker_detach(void);
static void fc_logicalrep_worker_cleanup(LogicalRepWorker *fc_worker);

static bool on_commit_launcher_wakeup = false;


/*
 * 加载订阅列表。
 *
 * 仅填充对工作者启动/停止功能有用的字段
 * 每个订阅。
 */
static List * fc_get_subscription_list(void)
{
	List	   *fc_res = NIL;
	Relation	fc_rel;
	TableScanDesc fc_scan;
	HeapTuple	fc_tup;
	MemoryContext fc_resultcxt;

	/* 这是我们将在其中分配输出数据的上下文 */
	fc_resultcxt = CurrentMemoryContext;

	/*
	 * 开始一个事务，以便我们可以访问 pg_database，并获取快照。
	 * 我们并不需要快照本身，但我们对它设置的 RecentGlobalXmin
	 * 的副作用很感兴趣。（这对读取堆页面的任何操作都是关键的，
	 * 因为 HOT 可能决定修剪它们，即使这个过程并没有试图修改任何元组。）
	 *
	 * FIXME: 这个评论不准确/代码有bug。一个未被推送/激活的快照
	 * 并不能可靠地防止 HOT 修剪（-> xmin 可能会在处理缓存无效时被清除）。
	 */
	StartTransactionCommand();
	(void) GetTransactionSnapshot();

	fc_rel = table_open(SubscriptionRelationId, AccessShareLock);
	fc_scan = table_beginscan_catalog(fc_rel, 0, NULL);

	while (HeapTupleIsValid(fc_tup = heap_getnext(fc_scan, ForwardScanDirection)))
	{
		Form_pg_subscription fc_subform = (Form_pg_subscription) GETSTRUCT(fc_tup);
		Subscription *fc_sub;
		MemoryContext fc_oldcxt;

		/*
		 * 在调用者的上下文中分配我们的结果，而不是在
		 * 事务中。我们在循环内部执行此操作，并在最后恢复原始
		 * 上下文，以便像 heap_getnext() 这样的泄露操作不会
		 * 在潜在的长期上下文中被调用。
		 */
		fc_oldcxt = MemoryContextSwitchTo(fc_resultcxt);

		fc_sub = (Subscription *) palloc0(sizeof(Subscription));
		fc_sub->oid = fc_subform->oid;
		fc_sub->dbid = fc_subform->subdbid;
		fc_sub->owner = fc_subform->subowner;
		fc_sub->enabled = fc_subform->subenabled;
		fc_sub->name = pstrdup(NameStr(fc_subform->subname));
		/* 我们不填充不感兴趣的字段。 */

		fc_res = lappend(fc_res, fc_sub);
		MemoryContextSwitchTo(fc_oldcxt);
	}

	table_endscan(fc_scan);
	table_close(fc_rel, AccessShareLock);

	CommitTransactionCommand();

	return fc_res;
}

/*
 * 等待后台工作者启动并附加到共享内存上下文。
 *
 * 这仅在工作者未能附加时需要进行共享内存的清理。
 */
static void fc_WaitForReplicationWorkerAttach(LogicalRepWorker *fc_worker,
							   uint16 fc_generation,
							   BackgroundWorkerHandle *fc_handle)
{
	BgwHandleStatus fc_status;
	int			fc_rc;

	for (;;)
	{
		pid_t		fc_pid;

		CHECK_FOR_INTERRUPTS();

		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);

		/* 工作者要么已死亡，要么已启动；无需做任何事情。 */
		if (!fc_worker->in_use || fc_worker->proc)
		{
			LWLockRelease(LogicalRepWorkerLock);
			return;
		}

		LWLockRelease(LogicalRepWorkerLock);

		/* 在附加之前检查工作者是否已死亡，并在之后进行清理。 */
		fc_status = GetBackgroundWorkerPid(fc_handle, &fc_pid);

		if (fc_status == BGWH_STOPPED)
		{
			LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
			/* 确保这确实是我们等待的工作者。 */
			if (fc_generation == fc_worker->generation)
				fc_logicalrep_worker_cleanup(fc_worker);
			LWLockRelease(LogicalRepWorkerLock);
			return;
		}

		/*
		 * 我们需要超时，因为我们通常不会通过拉增
		 * 来通知工作者附加的情况。但我们不期望等待太久。
		 */
		fc_rc = WaitLatch(MyLatch,
					   WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
					   10L, WAIT_EVENT_BGWORKER_STARTUP);

		if (fc_rc & WL_LATCH_SET)
		{
			ResetLatch(MyLatch);
			CHECK_FOR_INTERRUPTS();
		}
	}
}

/*
 * 遍历工作者数组并搜索与给定
 * 订阅 ID 和 relid 匹配的工作者。
 */
LogicalRepWorker *
logicalrep_worker_find(Oid fc_subid, Oid fc_relid, bool fc_only_running)
{
	int			fc_i;
	LogicalRepWorker *fc_res = NULL;

	Assert(LWLockHeldByMe(LogicalRepWorkerLock));

	/* 搜索给定订阅 ID 的附加工作者。 */
	for (fc_i = 0; fc_i < max_logical_replication_workers; fc_i++)
	{
		LogicalRepWorker *fc_w = &LogicalRepCtx->workers[fc_i];

		if (fc_w->in_use && fc_w->subid == fc_subid && fc_w->relid == fc_relid &&
			(!fc_only_running || fc_w->proc))
		{
			fc_res = fc_w;
			break;
		}
	}

	return fc_res;
}

/*
 * 类似于 logicalrep_worker_find()，但返回该
 * 订阅的所有工作者的列表，而不仅仅是一个。
 */
List * logicalrep_workers_find(Oid fc_subid, bool fc_only_running)
{
	int			fc_i;
	List	   *fc_res = NIL;

	Assert(LWLockHeldByMe(LogicalRepWorkerLock));

	/* 搜索给定订阅 ID 的附加工作者。 */
	for (fc_i = 0; fc_i < max_logical_replication_workers; fc_i++)
	{
		LogicalRepWorker *fc_w = &LogicalRepCtx->workers[fc_i];

		if (fc_w->in_use && fc_w->subid == fc_subid && (!fc_only_running || fc_w->proc))
			fc_res = lappend(fc_res, fc_w);
	}

	return fc_res;
}

/*
 * 如果可能，启动新的应用后台工作者。
 */
void logicalrep_worker_launch(Oid fc_dbid, Oid fc_subid, const char *fc_subname, Oid fc_userid,
						 Oid fc_relid)
{
	BackgroundWorker fc_bgw;
	BackgroundWorkerHandle *fc_bgw_handle;
	uint16		fc_generation;
	int			fc_i;
	int			fc_slot = 0;
	LogicalRepWorker *fc_worker = NULL;
	int			fc_nsyncworkers;
	TimestampTz fc_now;

	ereport(DEBUG1,
			(errmsg_internal("starting logical replication worker for subscription \"%s\"",
							 fc_subname)));

	/* 在初始启动消息之后报告此信息以保持一致性。 */
	if (max_replication_slots == 0)
		ereport(ERROR,
				(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
				 errmsg("cannot start logical replication workers when max_replication_slots = 0")));

	/*
	 * 我们需要在锁定下修改共享内存，以便
	 * 我们拥有一致的视图。
	 */
	LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);

retry:
	/* 查找未使用的工作者槽。 */
	for (fc_i = 0; fc_i < max_logical_replication_workers; fc_i++)
	{
		LogicalRepWorker *fc_w = &LogicalRepCtx->workers[fc_i];

		if (!fc_w->in_use)
		{
			fc_worker = fc_w;
			fc_slot = fc_i;
			break;
		}
	}

	fc_nsyncworkers = logicalrep_sync_worker_count(fc_subid);

	fc_now = GetCurrentTimestamp();

	/*
	 * 如果我们没有找到空闲槽，则尝试进行垃圾收集。
	 * 我们这样做的原因是因为如果某个工作者未能启动，
	 * 并且其父进程在等待时崩溃，则 in_use 状态从未被清除。
	 */
	if (fc_worker == NULL || fc_nsyncworkers >= max_sync_workers_per_subscription)
	{
		bool		fc_did_cleanup = false;

		for (fc_i = 0; fc_i < max_logical_replication_workers; fc_i++)
		{
			LogicalRepWorker *fc_w = &LogicalRepCtx->workers[fc_i];

			/*
			 * 如果工作者被标记为正在使用但未能及时附加，
			 * 则进行清理。
			 */
			if (fc_w->in_use && !fc_w->proc &&
				TimestampDifferenceExceeds(fc_w->launch_time, fc_now,
										   wal_receiver_timeout))
			{
				elog(WARNING,
					 "logical replication worker for subscription %u took too long to start; canceled",
					 fc_w->subid);

				fc_logicalrep_worker_cleanup(fc_w);
				fc_did_cleanup = true;
			}
		}

		if (fc_did_cleanup)
			goto retry;
	}

	/*
	 * 一旦达到每个订阅的同步工作者限制，我们不允许调用更多的同步工作者。
	 * 所以，静默返回，因为我们可能因为其它无害的竞争条件来到这里。
	 */
	if (OidIsValid(fc_relid) && fc_nsyncworkers >= max_sync_workers_per_subscription)
	{
		LWLockRelease(LogicalRepWorkerLock);
		return;
	}

	/*
	 * 然而，如果没有更多的空闲工作者槽，在退出之前通知用户。
	 */
	if (fc_worker == NULL)
	{
		LWLockRelease(LogicalRepWorkerLock);
		ereport(WARNING,
				(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
				 errmsg("out of logical replication worker slots"),
				 errhint("You might need to increase max_logical_replication_workers.")));
		return;
	}

	/* 准备工作者槽。 */
	fc_worker->launch_time = fc_now;
	fc_worker->in_use = true;
	fc_worker->generation++;
	fc_worker->proc = NULL;
	fc_worker->dbid = fc_dbid;
	fc_worker->userid = fc_userid;
	fc_worker->subid = fc_subid;
	fc_worker->relid = fc_relid;
	fc_worker->relstate = SUBREL_STATE_UNKNOWN;
	fc_worker->relstate_lsn = InvalidXLogRecPtr;
	fc_worker->stream_fileset = NULL;
	fc_worker->last_lsn = InvalidXLogRecPtr;
	TIMESTAMP_NOBEGIN(fc_worker->last_send_time);
	TIMESTAMP_NOBEGIN(fc_worker->last_recv_time);
	fc_worker->reply_lsn = InvalidXLogRecPtr;
	TIMESTAMP_NOBEGIN(fc_worker->reply_time);

	/* 在释放锁之前，记住所需的代数以便于将来的识别。 */
	fc_generation = fc_worker->generation;

	LWLockRelease(LogicalRepWorkerLock);

	/* 注册新的动态工作者。 */
	memset(&fc_bgw, 0, sizeof(fc_bgw));
	fc_bgw.bgw_flags = BGWORKER_SHMEM_ACCESS |
		BGWORKER_BACKEND_DATABASE_CONNECTION;
	fc_bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
	snprintf(fc_bgw.bgw_library_name, BGW_MAXLEN, "postgres");
	snprintf(fc_bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
	if (OidIsValid(fc_relid))
		snprintf(fc_bgw.bgw_name, BGW_MAXLEN,
				 "logical replication worker for subscription %u sync %u", fc_subid, fc_relid);
	else
		snprintf(fc_bgw.bgw_name, BGW_MAXLEN,
				 "logical replication worker for subscription %u", fc_subid);
	snprintf(fc_bgw.bgw_type, BGW_MAXLEN, "logical replication worker");

	fc_bgw.bgw_restart_time = BGW_NEVER_RESTART;
	fc_bgw.bgw_notify_pid = MyProcPid;
	fc_bgw.bgw_main_arg = Int32GetDatum(fc_slot);

	if (!RegisterDynamicBackgroundWorker(&fc_bgw, &fc_bgw_handle))
	{
		/* 启动工作者失败，因此清理工作者槽。 */
		LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
		Assert(fc_generation == fc_worker->generation);
		fc_logicalrep_worker_cleanup(fc_worker);
		LWLockRelease(LogicalRepWorkerLock);

		ereport(WARNING,
				(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
				 errmsg("out of background worker slots"),
				 errhint("You might need to increase max_worker_processes.")));
		return;
	}

	/* 现在等待它附加。 */
	fc_WaitForReplicationWorkerAttach(fc_worker, fc_generation, fc_bgw_handle);
}

/*
 * 停止指定 subid/relid 的逻辑复制工作者（如果有），并等待
 * 它从槽中分离。
 */
void logicalrep_worker_stop(Oid fc_subid, Oid fc_relid)
{
	LogicalRepWorker *fc_worker;
	uint16		fc_generation;

	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);

	fc_worker = logicalrep_worker_find(fc_subid, fc_relid, false);

	/* 没有工作者，没什么可做的。 */
	if (!fc_worker)
	{
		LWLockRelease(LogicalRepWorkerLock);
		return;
	}

	/*
	 * 记住我们的工作者是哪一代，以便检查我们看到的
	 * 是否仍然是同一个。
	 */
	fc_generation = fc_worker->generation;

	/*
	 * 如果我们找到一个工作者但它没有进程设置，那它还在
	 * 启动中；等待它启动完成，然后杀掉它。
	 */
	while (fc_worker->in_use && !fc_worker->proc)
	{
		int			fc_rc;

		LWLockRelease(LogicalRepWorkerLock);

		/* 等待一下——我们不希望等待太久。 */
		fc_rc = WaitLatch(MyLatch,
					   WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
					   10L, WAIT_EVENT_BGWORKER_STARTUP);

		if (fc_rc & WL_LATCH_SET)
		{
			ResetLatch(MyLatch);
			CHECK_FOR_INTERRUPTS();
		}

		/* 重新检查工作者状态。 */
		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);

		/*
		 * 检查工作者槽是否不再被使用，这将意味着
		 * 工作者已经退出，或者工作者的代数不同，这意味着另一个工作者
		 * 已经占用了这个槽。
		 */
		if (!fc_worker->in_use || fc_worker->generation != fc_generation)
		{
			LWLockRelease(LogicalRepWorkerLock);
			return;
		}

		/* 工作者已分配进程，因此它已经启动。 */
		if (fc_worker->proc)
			break;
	}

	/* 现在终止工作者... */
	kill(fc_worker->proc->pid, SIGTERM);

	/* ...并等待它死亡。 */
	for (;;)
	{
		int			fc_rc;

		/* 它不在了吗？ */
		if (!fc_worker->proc || fc_worker->generation != fc_generation)
			break;

		LWLockRelease(LogicalRepWorkerLock);

		/* 等待一下——我们不希望等待太久。 */
		fc_rc = WaitLatch(MyLatch,
					   WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
					   10L, WAIT_EVENT_BGWORKER_SHUTDOWN);

		if (fc_rc & WL_LATCH_SET)
		{
			ResetLatch(MyLatch);
			CHECK_FOR_INTERRUPTS();
		}

		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
	}

	LWLockRelease(LogicalRepWorkerLock);
}

/*
 * 使用闩锁唤醒指定 sub/rel 的任何逻辑复制工作者。
 */
void logicalrep_worker_wakeup(Oid fc_subid, Oid fc_relid)
{
	LogicalRepWorker *fc_worker;

	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);

	fc_worker = logicalrep_worker_find(fc_subid, fc_relid, true);

	if (fc_worker)
		logicalrep_worker_wakeup_ptr(fc_worker);

	LWLockRelease(LogicalRepWorkerLock);
}

/*
 * 使用闩锁唤醒指定的逻辑复制工作者。
 *
 * 调用者必须持有锁，否则 worker->proc 可能会在我们之下更改。
 */
void logicalrep_worker_wakeup_ptr(LogicalRepWorker *fc_worker)
{
	Assert(LWLockHeldByMe(LogicalRepWorkerLock));

	SetLatch(&fc_worker->proc->procLatch);
}

/*
 * 附加到一个槽。
 */
void logicalrep_worker_attach(int fc_slot)
{
	/* 阻止并发访问。 */
	LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);

	Assert(fc_slot >= 0 && fc_slot < max_logical_replication_workers);
	MyLogicalRepWorker = &LogicalRepCtx->workers[fc_slot];

	if (!MyLogicalRepWorker->in_use)
	{
		LWLockRelease(LogicalRepWorkerLock);
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("logical replication worker slot %d is empty, cannot attach",
						fc_slot)));
	}

	if (MyLogicalRepWorker->proc)
	{
		LWLockRelease(LogicalRepWorkerLock);
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("logical replication worker slot %d is already used by "
						"another worker, cannot attach", fc_slot)));
	}

	MyLogicalRepWorker->proc = MyProc;
	before_shmem_exit(fc_logicalrep_worker_onexit, (Datum) 0);

	LWLockRelease(LogicalRepWorkerLock);
}

/*
 * 分离工作者（清理工作者信息）。
 */
static void fc_logicalrep_worker_detach(void)
{
	/* 阻止并发访问。 */
	LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);

	fc_logicalrep_worker_cleanup(MyLogicalRepWorker);

	LWLockRelease(LogicalRepWorkerLock);
}

/*
 * 清理工作者信息。
 */
static void fc_logicalrep_worker_cleanup(LogicalRepWorker *fc_worker)
{
	Assert(LWLockHeldByMeInMode(LogicalRepWorkerLock, LW_EXCLUSIVE));

	fc_worker->in_use = false;
	fc_worker->proc = NULL;
	fc_worker->dbid = InvalidOid;
	fc_worker->userid = InvalidOid;
	fc_worker->subid = InvalidOid;
	fc_worker->relid = InvalidOid;
}

/*
 * 逻辑复制启动程序的清理函数。
 *
 * 在逻辑复制启动程序退出时调用。
 */
static void fc_logicalrep_launcher_onexit(int fc_code, Datum fc_arg)
{
	LogicalRepCtx->launcher_pid = 0;
}

/*
 * 清理函数。
 *
 * 在逻辑复制工作者退出时调用。
 */
static void fc_logicalrep_worker_onexit(int fc_code, Datum fc_arg)
{
	/* 从远程端优雅地断开连接。 */
	if (LogRepWorkerWalRcvConn)
		walrcv_disconnect(LogRepWorkerWalRcvConn);

	fc_logicalrep_worker_detach();

	/* 清理用于流式事务的文件集。 */
	if (MyLogicalRepWorker->stream_fileset != NULL)
		FileSetDeleteAll(MyLogicalRepWorker->stream_fileset);

	fc_ApplyLauncherWakeup();
}

/*
 * 计算为一个订阅注册的（不一定是正在运行的）同步工作者的数量。
 */
int logicalrep_sync_worker_count(Oid fc_subid)
{
	int			fc_i;
	int			fc_res = 0;

	Assert(LWLockHeldByMe(LogicalRepWorkerLock));

	/* 搜索给定订阅 ID 的附加工作者。 */
	for (fc_i = 0; fc_i < max_logical_replication_workers; fc_i++)
	{
		LogicalRepWorker *fc_w = &LogicalRepCtx->workers[fc_i];

		if (fc_w->subid == fc_subid && OidIsValid(fc_w->relid))
			fc_res++;
	}

	return fc_res;
}

/*
 * ApplyLauncherShmemSize
 *		计算逻辑复制启动程序所需的共享内存空间
 */
Size ApplyLauncherShmemSize(void)
{
	Size		fc_size;

	/*
	 * 需要固定的结构体和逻辑复制工作者的数组。
	 */
	fc_size = sizeof(LogicalRepCtxStruct);
	fc_size = MAXALIGN(fc_size);
	fc_size = add_size(fc_size, mul_size(max_logical_replication_workers,
								   sizeof(LogicalRepWorker)));
	return fc_size;
}

/*
 * ApplyLauncherRegister
 *		注册一个运行逻辑复制启动程序的后台工作者。
 */
void ApplyLauncherRegister(void)
{
	BackgroundWorker fc_bgw;

	if (max_logical_replication_workers == 0)
		return;

	memset(&fc_bgw, 0, sizeof(fc_bgw));
	fc_bgw.bgw_flags = BGWORKER_SHMEM_ACCESS |
		BGWORKER_BACKEND_DATABASE_CONNECTION;
	fc_bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
	snprintf(fc_bgw.bgw_library_name, BGW_MAXLEN, "postgres");
	snprintf(fc_bgw.bgw_function_name, BGW_MAXLEN, "ApplyLauncherMain");
	snprintf(fc_bgw.bgw_name, BGW_MAXLEN,
			 "logical replication launcher");
	snprintf(fc_bgw.bgw_type, BGW_MAXLEN,
			 "logical replication launcher");
	fc_bgw.bgw_restart_time = 5;
	fc_bgw.bgw_notify_pid = 0;
	fc_bgw.bgw_main_arg = (Datum) 0;

	RegisterBackgroundWorker(&fc_bgw);
}

/*
 * ApplyLauncherShmemInit
 *		分配和初始化逻辑复制启动程序的共享内存
 */
void ApplyLauncherShmemInit(void)
{
	bool		fc_found;

	LogicalRepCtx = (LogicalRepCtxStruct *)
		ShmemInitStruct("Logical Replication Launcher Data",
						ApplyLauncherShmemSize(),
						&fc_found);

	if (!fc_found)
	{
		int			fc_slot;

		memset(LogicalRepCtx, 0, ApplyLauncherShmemSize());

		/* 为每个工作者槽初始化内存和自旋锁。 */
		for (fc_slot = 0; fc_slot < max_logical_replication_workers; fc_slot++)
		{
			LogicalRepWorker *fc_worker = &LogicalRepCtx->workers[fc_slot];

			memset(fc_worker, 0, sizeof(LogicalRepWorker));
			SpinLockInit(&fc_worker->relmutex);
		}
	}
}

/*
 * 如果请求，则在提交时唤醒启动器。
 */
void AtEOXact_ApplyLauncher(bool fc_isCommit)
{
	if (fc_isCommit)
	{
		if (on_commit_launcher_wakeup)
			fc_ApplyLauncherWakeup();
	}

	on_commit_launcher_wakeup = false;
}

/*
 * 请求在事务提交时唤醒启动器。
 *
 * 这用于在当前事务提交时发送启动器信号以停止休眠并处理
 * 订阅。应在向 pg_subscription 目录添加新元组时使用。
*/
void ApplyLauncherWakeupAtCommit(void)
{
	if (!on_commit_launcher_wakeup)
		on_commit_launcher_wakeup = true;
}

static void fc_ApplyLauncherWakeup(void)
{
	if (LogicalRepCtx->launcher_pid != 0)
		kill(LogicalRepCtx->launcher_pid, SIGUSR1);
}

/*
 * 应用启动器进程的主循环。
 */
void ApplyLauncherMain(Datum fc_main_arg)
{
	TimestampTz fc_last_start_time = 0;

	ereport(DEBUG1,
			(errmsg_internal("logical replication launcher started")));

	before_shmem_exit(fc_logicalrep_launcher_onexit, (Datum) 0);

	Assert(LogicalRepCtx->launcher_pid == 0);
	LogicalRepCtx->launcher_pid = MyProcPid;

	/* 建立信号处理器。 */
	pqsignal(SIGHUP, SignalHandlerForConfigReload);
	pqsignal(SIGTERM, die);
	BackgroundWorkerUnblockSignals();

	/*
	 * 建立与钉住的目录的连接（我们只访问
	 * pg_subscription）。
	 */
	BackgroundWorkerInitializeConnection(NULL, NULL, 0);

	/* 进入主循环 */
	for (;;)
	{
		int			fc_rc;
		List	   *fc_sublist;
		ListCell   *fc_lc;
		MemoryContext fc_subctx;
		MemoryContext fc_oldctx;
		TimestampTz fc_now;
		long		fc_wait_time = DEFAULT_NAPTIME_PER_CYCLE;

		CHECK_FOR_INTERRUPTS();

		fc_now = GetCurrentTimestamp();

		/* 将启动重试限制为一次 wal_retrieve_retry_interval */
		if (TimestampDifferenceExceeds(fc_last_start_time, fc_now,
									   wal_retrieve_retry_interval))
		{
			/* 使用临时上下文处理数据库列表和工作信息。 */
			fc_subctx = AllocSetContextCreate(TopMemoryContext,
										   "Logical Replication Launcher sublist",
										   ALLOCSET_DEFAULT_SIZES);
			fc_oldctx = MemoryContextSwitchTo(fc_subctx);

			/* 搜索要启动或停止的订阅。 */
			fc_sublist = fc_get_subscription_list();

			/* 为启用的订阅启动缺失的工作者。 */
			foreach(fc_lc, fc_sublist)
			{
				Subscription *fc_sub = (Subscription *) lfirst(fc_lc);
				LogicalRepWorker *fc_w;

				if (!fc_sub->enabled)
					continue;

				LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
				fc_w = logicalrep_worker_find(fc_sub->oid, InvalidOid, false);
				LWLockRelease(LogicalRepWorkerLock);

				if (fc_w == NULL)
				{
					fc_last_start_time = fc_now;
					fc_wait_time = wal_retrieve_retry_interval;

					logicalrep_worker_launch(fc_sub->dbid, fc_sub->oid, fc_sub->name,
											 fc_sub->owner, InvalidOid);
				}
			}

			/* 切换回原始内存上下文。 */
			MemoryContextSwitchTo(fc_oldctx);
			/* 清理临时内存。 */
			MemoryContextDelete(fc_subctx);
		}
		else
		{
			/*
			 * 上一个周期中的等待在最后一个工作者启动后的
			 * wal_retrieve_retry_interval 内被中断，这通常意味着
			 * 工作者崩溃，因此我们应该在 wal_retrieve_retry_interval
			 * 中再尝试一次。
			 */
			fc_wait_time = wal_retrieve_retry_interval;
		}

		/* 等待更多工作。 */
		fc_rc = WaitLatch(MyLatch,
					   WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
					   fc_wait_time,
					   WAIT_EVENT_LOGICAL_LAUNCHER_MAIN);

		if (fc_rc & WL_LATCH_SET)
		{
			ResetLatch(MyLatch);
			CHECK_FOR_INTERRUPTS();
		}

		if (ConfigReloadPending)
		{
			ConfigReloadPending = false;
			ProcessConfigFile(PGC_SIGHUP);
		}
	}

	/* 无法到达 */
}

/*
 * 当前进程是逻辑复制启动器吗？
 */
bool IsLogicalLauncher(void)
{
	return LogicalRepCtx->launcher_pid == MyProcPid;
}

/*
 * 返回订阅的状态。
 */
Datum pg_stat_get_subscription(PG_FUNCTION_ARGS)
{
#define PG_STAT_GET_SUBSCRIPTION_COLS	8
	Oid			fc_subid = PG_ARGISNULL(0) ? InvalidOid : PG_GETARG_OID(0);
	int			fc_i;
	ReturnSetInfo *fc_rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;

	InitMaterializedSRF(fcinfo, 0);

	/* 确保我们获得工作者的一致视图。 */
	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);

	for (fc_i = 0; fc_i < max_logical_replication_workers; fc_i++)
	{
		/* 对于每一行 */
		Datum		fc_values[PG_STAT_GET_SUBSCRIPTION_COLS];
		bool		fc_nulls[PG_STAT_GET_SUBSCRIPTION_COLS];
		int			fc_worker_pid;
		LogicalRepWorker fc_worker;

		memcpy(&fc_worker, &LogicalRepCtx->workers[fc_i],
			   sizeof(LogicalRepWorker));
		if (!fc_worker.proc || !IsBackendPid(fc_worker.proc->pid))
			continue;

		if (OidIsValid(fc_subid) && fc_worker.subid != fc_subid)
			continue;

		fc_worker_pid = fc_worker.proc->pid;

		MemSet(fc_values, 0, sizeof(fc_values));
		MemSet(fc_nulls, 0, sizeof(fc_nulls));

		fc_values[0] = ObjectIdGetDatum(fc_worker.subid);
		if (OidIsValid(fc_worker.relid))
			fc_values[1] = ObjectIdGetDatum(fc_worker.relid);
		else
			fc_nulls[1] = true;
		fc_values[2] = Int32GetDatum(fc_worker_pid);
		if (XLogRecPtrIsInvalid(fc_worker.last_lsn))
			fc_nulls[3] = true;
		else
			fc_values[3] = LSNGetDatum(fc_worker.last_lsn);
		if (fc_worker.last_send_time == 0)
			fc_nulls[4] = true;
		else
			fc_values[4] = TimestampTzGetDatum(fc_worker.last_send_time);
		if (fc_worker.last_recv_time == 0)
			fc_nulls[5] = true;
		else
			fc_values[5] = TimestampTzGetDatum(fc_worker.last_recv_time);
		if (XLogRecPtrIsInvalid(fc_worker.reply_lsn))
			fc_nulls[6] = true;
		else
			fc_values[6] = LSNGetDatum(fc_worker.reply_lsn);
		if (fc_worker.reply_time == 0)
			fc_nulls[7] = true;
		else
			fc_values[7] = TimestampTzGetDatum(fc_worker.reply_time);

		tuplestore_putvalues(fc_rsinfo->setResult, fc_rsinfo->setDesc,
							 fc_values, fc_nulls);

		/*
		 * 如果只请求了一个订阅，并且我们找到了它，
		 * 则退出。
		 */
		if (OidIsValid(fc_subid))
			break;
	}

	LWLockRelease(LogicalRepWorkerLock);

	return (Datum) 0;
}
